package cn.felord.debezium.engine;

import cn.felord.debezium.annotation.Executor;
import cn.felord.debezium.handle.columnHandle.ColumnHandleExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.connect.data.Struct;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.annotation.Annotation;
import java.util.Map;
import java.util.Set;

@Configuration
@Slf4j
public class OperationColumnHandleEngine {

    @Resource
    private ApplicationContext applicationContext;
    public void handle(String table, Struct before, Struct after) {
        Map<String, ColumnHandleExecutor> beansOfType = applicationContext.getBeansOfType(ColumnHandleExecutor.class);
        beansOfType.entrySet().forEach(x->{
            ColumnHandleExecutor value = x.getValue();
            Executor annotation = value.getClass().getAnnotation(Executor.class);
            if (annotation.tableName().equals(table)){
                value.handleUpdate(before,after);
                return;
            }
        });
    }
    public void handle(String table,String operation,Map<String, Object> payload ) {
        Map<String, ColumnHandleExecutor> beansOfType = applicationContext.getBeansOfType(ColumnHandleExecutor.class);
        beansOfType.entrySet().forEach(x->{
            ColumnHandleExecutor value = x.getValue();
            Executor annotation = value.getClass().getAnnotation(Executor.class);
            if (annotation.tableName().equals(table)){
                if (operation.equalsIgnoreCase("CREATE")){
                    value.handleInsert(payload);
                }
                if (operation.equalsIgnoreCase("DELETE")){
                    value.handleDelete(payload);
                }
                return;
            }
        });
    }
}
